package com.shanks.framework.mq.service.impl;

import cn.hutool.core.util.IdUtil;
import com.shanks.framework.mq.common.util.MQUtil;
import com.shanks.framework.mq.enums.MessageRecordStatusEnum;
import com.shanks.framework.mq.helper.MqHelper;
import com.shanks.framework.mq.model.MessageRecordPO;
import com.shanks.framework.mq.queue.MessageQueueManager;
import com.shanks.framework.mq.repository.MessageRecordRepository;
import com.shanks.framework.mq.service.MessageQueueService;
import com.shanks.framework.mq.service.ProducerService;
import jakarta.annotation.Resource;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.stereotype.Service;
import org.springframework.transaction.support.TransactionSynchronizationAdapter;
import org.springframework.transaction.support.TransactionSynchronizationManager;

import java.util.Arrays;
import java.util.Date;
import java.util.List;
import java.util.stream.Collectors;

/**
 * 本地消息表服务实现类
 */
@Service
public class MessageQueueServiceImpl implements MessageQueueService {

    private static Logger logger = LoggerFactory.getLogger(MessageQueueServiceImpl.class);

    @Value("${rocketmq.default-producer.producerKey}")
    private String defaultProducerKey;

    /**
     * 消息插入检查次数
     */
    @Value("${mq-message.ack-checked-limit:2}")
    private int messageAckCheckedLimit;

    @Resource
    private MqHelper mqHelper;

    @Resource
    private ProducerService producerService;

    @Resource
    private MessageQueueManager messageQueueManager;

    @Resource
    private MessageRecordRepository messageRecordRepository;


    @Override
    public void insertAndOfferMessage(MessageRecordPO messageRecordPO) {
        int retry = 0;
        // 批量插入并入队
        while (retry <= this.messageAckCheckedLimit) {
            try {
                // 填充字段
                buildMessageRecordPO(messageRecordPO);

                // 插入数据库
                messageRecordRepository.insert(messageRecordPO);

                // 入队
                offerBatch(Arrays.asList(messageRecordPO), MessageRecordStatusEnum.PENDING);
                return;  // 成功则退出方法

            } catch (DuplicateKeyException e) {
                // 处理主键冲突异常
                if (isDuplicateKeyException(e)) {
                    handleDuplicateKeyException(messageRecordPO, retry, e);
                    retry++;
                    continue;
                }
                // 其他类型的 DuplicateKeyException 异常
                logErrorAndThrow("发送MQ消息异常(主键冲突)", messageRecordPO, retry, e);
            } catch (Exception e) {
                // 捕获其他异常
                logErrorAndThrow("插入MQ消息异常", messageRecordPO, retry, e);
            }
        }

        // 如果重试次数超过限制，抛出异常
        logErrorAndThrow("重试次数超限", messageRecordPO, retry, null);
    }

    /**
     * 判断是否为主键冲突异常
     */
    private boolean isDuplicateKeyException(DuplicateKeyException e) {
        return e.getMessage() != null && e.getMessage().contains("message_record.PRIMARY");
    }

    /**
     * 处理主键冲突的异常
     */
    private void handleDuplicateKeyException(MessageRecordPO messageRecordPO, int retry, Exception e) {
        if (retry < this.messageAckCheckedLimit) {
            logger.warn("发送MQ消息异常(主键冲突), topic: {}, msgKey:{}, msgTags:{}, msgBizType: {}, msgBizUuid: {}, 重试次数: {}",
                    messageRecordPO.getMessageTopic(), messageRecordPO.getMessageUuid(), messageRecordPO.getMessageTag(),
                    messageRecordPO.getBizType(), messageRecordPO.getBizUuid(), retry, e);
        } else {
            logger.error("发送MQ消息异常(主键冲突), topic: {}, msgKey:{}, msgTags:{}, msgBizType: {}, msgBizUuid: {}, 重试次数: {}",
                    messageRecordPO.getMessageTopic(), messageRecordPO.getMessageUuid(), messageRecordPO.getMessageTag(),
                    messageRecordPO.getBizType(), messageRecordPO.getBizUuid(), retry, e);
        }
    }

    /**
     * 记录异常并抛出
     */
    private void logErrorAndThrow(String message, MessageRecordPO messageRecordPO, int retry, Exception e) {
        if (e != null) {
            // 记录异常日志
            logger.error("{}: topic: {}, msgKey:{}, msgTags:{}, msgBizType: {}, msgBizUuid: {}, 重试次数: {}",
                    message, messageRecordPO.getMessageTopic(), messageRecordPO.getMessageUuid(), messageRecordPO.getMessageTag(),
                    messageRecordPO.getBizType(), messageRecordPO.getBizUuid(), retry, e);

            // 包装成 RuntimeException 抛出，避免异常类型不匹配问题
            throw new RuntimeException(message, e);
        } else {
            // 无异常时记录普通日志
            logger.error("{}: topic: {}, msgKey:{}, msgTags:{}, msgBizType: {}, msgBizUuid: {}, 重试次数: {}",
                    message, messageRecordPO.getMessageTopic(), messageRecordPO.getMessageUuid(), messageRecordPO.getMessageTag(),
                    messageRecordPO.getBizType(), messageRecordPO.getBizUuid(), retry);

            // 抛出一个新的 RuntimeException
            throw new RuntimeException(message);
        }
    }

    @Override
    public void insertAndOfferBatchMessages(List<MessageRecordPO> messageRecordPOList) {
        int retry = 0;
        while (retry <= this.messageAckCheckedLimit) {
            try {
                // 填充每一条记录的字段
                messageRecordPOList.forEach(this::buildMessageRecordPO);

                // 批量插入消息记录
                messageRecordRepository.insertBatch(messageRecordPOList);

                // 批量入队
                offerBatch(messageRecordPOList, MessageRecordStatusEnum.PENDING);
                return;  // 批量处理成功则退出方法

            } catch (DuplicateKeyException e) {
                // 处理主键冲突异常
                if (isDuplicateKeyException(e)) {
                    handleDuplicateKeyExceptionForBatch(retry, e);
                    retry++;
                    continue;
                }
                // 其他类型的 DuplicateKeyException 异常
                logErrorAndThrow("批量发送MQ消息异常(主键冲突)", messageRecordPOList, retry, e);
            } catch (Exception e) {
                // 捕获其他异常
                logErrorAndThrow("批量插入MQ消息异常", messageRecordPOList, retry, e);
            }
        }

        // 如果重试次数超过限制，抛出异常
        logErrorAndThrow("批量重试次数超限", messageRecordPOList, retry, null);
    }

    /**
     * 处理批量主键冲突的异常
     */
    private void handleDuplicateKeyExceptionForBatch(int retry, Exception e) {
        if (retry < this.messageAckCheckedLimit) {
            logger.warn("批量发送MQ消息异常(主键冲突), 重试次数: {}", retry, e);
        } else {
            logger.error("批量发送MQ消息异常(主键冲突), 重试次数: {}", retry, e);
        }
    }

    /**
     * 批量记录错误并抛出异常
     */
    private void logErrorAndThrow(String message, List<MessageRecordPO> messageRecordPOList, int retry, Exception e) {
        if (e != null) {
            logger.error("{}: 批量消息记录数: {}, 重试次数: {}", message, messageRecordPOList.size(), retry, e);
            // 包装成 RuntimeException 抛出，避免异常类型不匹配问题
            throw new RuntimeException(message, e);
        } else {
            logger.error("{}: 批量消息记录数: {}, 重试次数: {}", message, messageRecordPOList.size(), retry);

            // 抛出一个新的 RuntimeException
            throw new RuntimeException(message);
        }
    }

    private void offerBatch(final List<MessageRecordPO> messageRecordPOList, final MessageRecordStatusEnum messageRecordStatus) {
        // 判断当前是否有事务存在
        if (TransactionSynchronizationManager.isActualTransactionActive()) {
            // 如果事务存在，注册事务同步器，在事务提交后执行操作
            TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
                @Override
                public void afterCommit() {
                    try {
                        // 事务提交完成后的操作，压入消息队列
                        logger.debug("事务提交成功，准备将消息压入队列，消息ID集合: {}",
                                messageRecordPOList.stream().map(MessageRecordPO::getMessageId).collect(Collectors.toList()));
                        messageQueueManager.offerBatchMessages(messageRecordPOList, messageRecordStatus);
                    } catch (Exception exception) {
                        // 异常日志，确保即使出现问题也能记录
                        logger.error("事务提交后压入消息队列失败，消息ID集合: {}",
                                messageRecordPOList.stream().map(MessageRecordPO::getMessageId).collect(Collectors.toList()), exception);
                    }
                }
            });
        } else {
            // 如果没有事务，直接将消息压入队列
            try {
                logger.debug("没有事务，直接将消息压入队列，消息ID集合: {}",
                        messageRecordPOList.stream().map(MessageRecordPO::getMessageId).collect(Collectors.toList()));
                messageQueueManager.offerBatchMessages(messageRecordPOList, messageRecordStatus);
            } catch (Exception exception) {
                // 异常日志
                logger.error("没有事务时压入消息队列失败，消息ID集合: {}",
                        messageRecordPOList.stream().map(MessageRecordPO::getMessageId).collect(Collectors.toList()), exception);
            }
        }
    }


    /**
     * 填充生成本地消息部分字段
     *
     * @param messageRecordPO
     */
    private void buildMessageRecordPO(MessageRecordPO messageRecordPO) {
        messageRecordPO.setMessageId(IdUtil.getSnowflakeNextId());
        // uuid topic_json md5 唯一
        messageRecordPO.setMessageUuid(MQUtil.generateMessageUuid(messageRecordPO.getMessageTopic(), messageRecordPO.getMessageJson()));
        messageRecordPO.setCreateTime(new Date());
        messageRecordPO.setRetryCount(0);
        // 初始状态
        messageRecordPO.setMessageRecordStatus(MessageRecordStatusEnum.PENDING);
        // 生产者
        if (StringUtils.isBlank(messageRecordPO.getProducerSource())) {
            messageRecordPO.setProducerSource(mqHelper.getInstanceName());
        }
        if (StringUtils.isBlank(messageRecordPO.getProducerKey())) {
            messageRecordPO.setProducerKey(this.defaultProducerKey);
        }
        messageRecordPO.setMqZone(producerService.getZoneByProducerKeyQuietly(messageRecordPO.getProducerKey()));
    }
}
